package process

import (
	"os"
	"path/filepath"
	"io/ioutil"
	"encoding/json"
	"bytes"
	"zipfile"
	"kafka"
)

type fileprocess struct {
	AbsPath string
	PerPath	string
	State state
}

func NewFileProcess(abspath string)*fileprocess{
	absdir := filepath.Dir(abspath)
	basename := filepath.Base(abspath)

	fp := new(fileprocess)
	fp.AbsPath = abspath
	fp.PerPath = filepath.Join(absdir,".notify",basename+".json")
	fp.State = state{
		AbsPath:fp.AbsPath,
		ProcessedLine:0,
		Done:false,
	}

	return fp
}


func (fp *fileprocess)FileProcess(k *kafka.Kafka) error{
	fp.loadState()
	if fp.done()==true{
		os.Remove(fp.AbsPath)
		return nil
	}
	breakFlag := false
	if lines,err := zipfile.ReadLines(fp.AbsPath);err == nil{
		for ln := range lines{
			if ln > fp.State.ProcessedLine{
				wfid := os.Getenv("GWWF_WFID")
				data := wfid+","+lines[ln]
				err := k.Send(data)
				if err !=nil{
					breakFlag = true
					break
				}
				fp.State.ProcessedLine = ln
			}
		}
		if breakFlag{
			fp.State.Done = false
		}else {
			fp.State.Done = true
			os.Remove(fp.AbsPath)
		}
	}
	fp.saveState()

	return nil
}

func (fp fileprocess)done()bool{
	return fp.State.Done
}

func (fp *fileprocess)loadState() error{

	if _,err := os.Stat(fp.PerPath); err==nil {
		data,err := ioutil.ReadFile(fp.PerPath)
		//fmt.Println(string(data))
		if err != nil{
			return err
		}
		json.Unmarshal(data, &fp.State)
	}

	return nil
}

func (fp fileprocess)saveState()error{
	data,err := json.Marshal(fp.State)
	var format bytes.Buffer

	json.Indent(&format,data,"","  ")
	if err != nil{
		return err
	}
	ioutil.WriteFile(fp.PerPath,format.Bytes(),os.ModePerm)
	return nil
}
